package org.dyndns.opendemogroup.hadoopoptimizer;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.dyndns.opendemogroup.optimizer.GenericGenerationalGeneticAlgorithm;
import org.dyndns.opendemogroup.optimizer.OptimizationProfile;
import org.dyndns.opendemogroup.optimizer.RunResult;

public class DistributedRunMapper extends MapReduceBase implements Mapper
{

	private final static Text outputKey = new Text ( );

	private final static Text outputValue = new Text ( );

	/**
	 * Emits a &lt;bestFitness, genome&gt; pair representing the results of the
	 * optimizer run.
	 * 
	 * @param key
	 *        Represents the file offset at which <i>values</i> was read.
	 * @param values
	 *        Represents the OptimizationProfile serialized as XML. 
	 * @see Mapper#map(WritableComparable, Writable, OutputCollector, Reporter)
	 */
	@SuppressWarnings ( "unchecked" )
	public void map ( WritableComparable key, Writable values,
			OutputCollector output, Reporter reporter ) throws IOException
	{
		String configurationXml = values.toString ( );
		OptimizationProfile profile =
			OptimizationProfile.fromXmlString ( configurationXml );
		GenericGenerationalGeneticAlgorithm ggga =
			new GenericGenerationalGeneticAlgorithm ( profile );
		RunResult[] results = ggga.execute ( );
		for ( int i = 0; i < results.length; i++ )
		{
			outputKey.set ( Double.toString ( results[i].BestFitness ) );
			outputValue.set ( results[i].BestMember.toString ( ) );
			output.collect ( outputKey, outputValue );
		}
	}

}
